package crossjob;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Set;

/**
 * Created by zhaopeng on 2/7/17.
 */
public class CrossJobReducer extends Reducer<Text, Text, Text, Text> {
    @Override
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
        Set tags = new HashSet<Integer>();
        boolean isCross = false;
        for (Text tmp : values) {
            String strTag = tmp.toString();
            String[] nums = strTag.split(",");
            for (int i = 0; i < nums.length; i++) {
                try {
                    Integer a = Integer.parseInt(nums[i]);
                    if (a.equals(0)) {
                        isCross = true;
                    } else {
                        tags.add(a);
                    }
                } catch (NumberFormatException e) {

                }
            } // end of inner for
        }// end of outer for
        if (isCross && tags.size() > 0) {
            String result = "";
            Iterator it = tags.iterator();
            result += it.next().toString();
            for (; it.hasNext(); )
                result += "," + it.next().toString();
            context.write(key, new Text(result));
        }

    }
}
